package com.gt.ln

import com.gt.SCUtil
import org.apache.commons.lang3.StringUtils
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

object Spark_hot_product_01 {

  def main(args: Array[String]): Unit = {
    val sc: SparkContext = SCUtil.createLocalSc()

    //1. 获取数据
    val rdd: RDD[String] = sc.textFile("data/user_visit_action.txt")

    //2.  转换数据格式为(id,(clickCount,bookCount,payCount))
    //2.1 点击类 品类排行
    //      //val clickCategoryId = arr(6) //点击类 品类id
    //      //val bookCategoryId = arr(8) //下单类 品类id a-b-c
    //      //val payCategoryId = arr(10) //支付类 品类id a-b-c
    val clickRdd: RDD[(String, Int)] = rdd.map(line => line.split("_"))
      .filter(arr => StringUtils.isNotBlank(arr(6)) && !arr(6).equals("-1"))
      .map(arr => (arr(6), 1))
      .reduceByKey(_+_)
      .sortBy(data => data._2)

    //2.2 下单类 品类排行
    val bookRdd: RDD[(String, Int)] = rdd.map(line => line.split("_"))
      .filter(arr => StringUtils.isNotBlank(arr(8)))
      .flatMap(arr => arr(8).split(","))
      .map(id => (id, 1))
      .reduceByKey(_ + _)
      .sortBy(data => data._2)


    //2.3 支付类 品类排行
    val payRdd: RDD[(String, Int)] = rdd.map(line => line.split("_"))
      .filter(arr => StringUtils.isNotBlank(arr(10)))
      .flatMap(arr => arr(10).split(","))
      .map(id => (id, 1))
      .reduceByKey(_ + _)
      .sortBy(data => data._2)

    val rdd1: RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Int]))] = clickRdd.cogroup(bookRdd, payRdd)

    val resultRdd: RDD[(String, (Int, Int, Int))] = rdd1.map(data => {
      val clickIt: Iterator[Int] = data._2._1.iterator
      val bookIt: Iterator[Int] = data._2._2.iterator
      val payIt: Iterator[Int] = data._2._3.iterator

      var clickCount = 0
      var bookCount = 0
      var payCount = 0

      if (clickIt.hasNext) {
        clickCount = clickIt.next()
      }
      if (bookIt.hasNext) {
        bookCount = bookIt.next()
      }
      if (payIt.hasNext) {
        payCount = payIt.next()
      }
      (data._1, (clickCount, bookCount, payCount))
    }).sortBy(data => data._2,false)

    resultRdd.collect().foreach(println)

    sc.stop()
  }

}
